#ifndef __M__CHANNEL_H_
#define __M__CHANNEL_H_

#include "../mqCommon/Helper.hpp"
#include "../mqCommon/Logger.hpp"
#include "../mqCommon/threadPool.hpp"

#include "consumer.hpp"
#include "virtual_host.hpp"
#include "route.hpp"

#include "../mqCommon/proto/mq_message.pb.h"
#include "../mqCommon/proto/mq_proto.pb.h"
#include "../third/muduo/include/muduo/net/TcpConnection.h"
#include "../third/muduo/proto/codec.h"
#include "../third/muduo/proto/dispatcher.h"

#include <iostream>
#include <unordered_map>
#include <memory>
#include <mutex>
#include <vector>
#include <functional>

using namespace mq_message;
using namespace mq;
using namespace std;
using namespace muduo::net;

namespace mq
{
    using ProtobufCodecPtr = shared_ptr<ProtobufCodec>; // 协议处理器

    // 信道提高网络服务，就需要定制应用层协议，以请求的方式进行处理
    // 请求/响应的智能指针类型重定义
    using openChannelRequestPtr = shared_ptr<openChannelRequest>;
    using closeChannelRequestPtr = shared_ptr<closeChannelRequest>;
    using declareExchangeRequestPtr = shared_ptr<declareExchangeRequest>;
    using deleteExchangeRequestPtr = shared_ptr<deleteExchangeRequest>;
    using declareQueueRequestPtr = shared_ptr<declareQueueRequest>;
    using deleteQueueRequestPtr = shared_ptr<deleteQueueRequest>;
    using queueBindRequestPtr = shared_ptr<queueBindRequest>;
    using queueUnBindRequestPtr = shared_ptr<queueUnBindRequest>;
    using basicConsumeRequestPtr = shared_ptr<basicConsumeRequest>;
    using basicCancelRequestPtr = shared_ptr<basicCancelRequest>;
    using basicPublishRequestPtr = shared_ptr<basicPublishRequest>;
    using basicAckRequestPtr = shared_ptr<basicAckRequest>;

    /*  信道
        信道Channel:
            1.连接Connection的更细粒度划分，提高资源利用率
            2.一个连接对应多个信道，不同信道使用同一连接但互相独立

        元素：
            1.信道ID
            2.信道关联的消费者
            3.信道关联的连接
            4.消费者管理句柄
            5.虚拟机句柄
            6.工作线程池句柄
            7.protobuf协议处理句柄

        操作：
            1.声明/删除交换机
            2.声明/删除队列
            3.绑定/解绑
            4.发布消息/确认消息
            5.订阅队列/取消订阅
    */
    // 信道类
    class Channel
    {
    public:
        using ptr = shared_ptr<Channel>;
        Channel(const string &cid,
                const ConsumerManager::ptr &cmp,
                const TcpConnectionPtr &conn,
                const ProtobufCodecPtr &codec,
                const VirtualHost::ptr &host,
                const ThreadPool::ptr &tpool)
            : _cid(cid), _cmp(cmp), _conn(conn), _codec(codec), _host(host), _tpool(tpool) {}
        ~Channel()
        {
            if (_consumer.get() != nullptr) // 信道关闭时，将关联的消费者删除
            {
                _cmp->remove(_consumer->_qname, _consumer->_tag);
            }
        }
        void declareExchange(const declareExchangeRequestPtr &req)
        {
            string ename = req->exchange_name();
            ExchangeType type = req->exchange_type();
            bool durable = req->durable();
            bool auto_delete = req->auto_delete();
            Map args = req->args();
            bool ret = _host->declareExchange(ename, type, durable, auto_delete, args);

            basicResponse(req->rid(), req->cid(), ret);
        }
        void deleteExchange(const deleteExchangeRequestPtr &req)
        {
            string del_ename = req->exchange_name();
            _host->deleteExchange(del_ename);
            basicResponse(req->rid(), req->cid(), true);
        }
        void declareMsgQueue(const declareQueueRequestPtr &req)
        {
            string qname = req->queue_name();
            bool exclusive = req->exclusive();
            bool durable = req->durable();
            bool auto_delete = req->auto_delete();
            Map args = req->args();
            bool ret = _host->declareMsgQueue(qname, exclusive, durable, auto_delete, args);
            _cmp->initQueueConsumer(qname); // 队列声明：队列/队列消息/队列消费者

            basicResponse(req->rid(), req->cid(), ret);
        }
        void deleteMsgQueue(const deleteQueueRequestPtr &req)
        {
            string qname = req->queue_name();
            _cmp->destroyQueueConsumer(qname);
            _host->deleteMsgQueue(qname); // 队列的删除：队列/队列绑定/队列消息/队列消费者

            basicResponse(req->rid(), req->cid(), true);
        }
        void bind(const queueBindRequestPtr &req)
        {
            string ename = req->exchange_name();
            string qname = req->queue_name();
            string binding_key = req->binding_key();
            bool ret = _host->Bind(ename, qname, binding_key);

            basicResponse(req->rid(), req->cid(), ret);
        }
        void unBind(const queueUnBindRequestPtr &req)
        {
            string ename = req->exchange_name();
            string qname = req->queue_name();
            _host->unBind(ename, qname);

            basicResponse(req->rid(), req->cid(), true);
        }
        void basicPublish(const basicPublishRequestPtr &req) // 消息发布：指定交换机->绑定队列->待推送链表->制作任务，推送到订阅者
        {
            // 1.判断交换机是否存在
            string ename = req->exchange_name();
            bool exists = _host->existExchange(ename);
            if (!exists) // 交换机不存在
            {
                DLOG("消息发布失败，交换机不存在：%s", ename.c_str());
                basicResponse(req->rid(), req->cid(), false);
                return;
            }

            // 2.获取交换机绑定
            MsgQueueBindingMap mqbm = _host->getExchangeBindings(ename);

            // 3.遍历交换路由，将消息交给符合匹配规则的绑定队列
            // 交换路由需要：交换机类型，rkey和bkey；并根据是否匹配，交给指定队列

            // 交换机类型
            Exchange::ptr exchange = _host->selectExchange(ename); // 交换机
            ExchangeType type = exchange->_type;
            // rkey
            bool has_properties = req->has_properties(); // 用户可能为了广播，不传properties
            string routing_key;
            if (has_properties)
                routing_key = req->properties().routing_key();

            for (auto &it : mqbm) // 遍历所有绑定队列
            {
                Binding::ptr binding = it.second;
                string qname = binding->_qname; // 绑定到该交换机的队列
                string binding_key = binding->_binding_key;

                bool match = Router::route(type, routing_key, binding_key);
                if (match) // 匹配成功
                {
                    BasicProperties *bp = has_properties ? req->mutable_properties() : nullptr;
                    _host->basicPublish(qname, bp, req->body()); // 消息发布到该队列中（待推送链表）

                    auto task = std::bind(&Channel::consume, this, qname); // 此时，该队列中一定有待推送消息
                    _tpool->PushTask(task);                                // 制作推送（消费）任务，加入任务池，由工作线程去执行
                }
            }
            basicResponse(req->rid(), req->cid(), true);
        }
        void basicAck(const basicAckRequestPtr &req) // 消息应答
        {
            _host->basicAck(req->queue_name(), req->message_id());
            basicResponse(req->rid(), req->cid(), true);
        }
        void basicConsume(const basicConsumeRequestPtr &req) // 订阅队列
        {
            string qname = req->queue_name();  // 订阅队列名称
            string ctag = req->consumer_tag(); // 消费者标识
            bool auto_ack = req->auto_ack();   // 自动应答标志

            // 创建消费者，放入指定队列消费者结构，进行管理
            auto cb = std::bind(&Channel::consumerCallback, this, std::placeholders::_1,
                                std::placeholders::_2, std::placeholders::_3);
            // 订阅队列 = 创建消费者 = 该信道就是为该消费者服务的
            _consumer = _cmp->create(qname, ctag, auto_ack, cb);
            if (_consumer == Consumer::ptr()) // 找不到指定队列
            {
                basicResponse(req->rid(), req->cid(), false);
                return;
            }
            basicResponse(req->rid(), req->cid(), true);
        }
        void basicCancel(const basicCancelRequestPtr &req) // 取消订阅
        {
            string qname = req->queue_name();  // 取消订阅队列名称
            string ctag = req->consumer_tag(); // 消费者标识

            // 从指定队列消费者结构，移除该消费者
            _cmp->remove(qname, ctag);
            // 将该信道关联的消费者重置
            _consumer.reset();
            basicResponse(req->rid(), req->cid(), true);
        }

    private:
        void basicResponse(const string &rid, const string &cid, bool ok)
        {
            // 构建响应并发回
            basicCommonResponse resp;
            resp.set_rid(rid);
            resp.set_cid(cid);
            resp.set_ok(ok);
            if(_conn.get()==nullptr){
                cout<<"_conn=nullptr"<<endl;
                return;
            }
            if(!_conn->connected()){
                cout<<"unconnected"<<endl;
                return;
            }
            _codec->send(_conn, resp);
        }
        void consume(const string &qname) // 指定队列中有消息，进行消费（推送）:该函数与当前信道无关，只是一个工作线程的执行任务函数
        {
            // 1.从该队列的订阅者中，按序取出一个消费者
            Consumer::ptr consumer = _cmp->choose(qname);
            if (consumer == Consumer::ptr())
            {
                DLOG("推送(消费)失败，该队列无消费者可推送：%s", qname.c_str());
                return;
            }
            // 2.从该队列将消息拿出
            m_ptr msgp = _host->basicConsume(qname);
            if (msgp == m_ptr())
            {
                DLOG("推送(消费)失败，该队列无消息可推送：%s", qname.c_str());
                return;
            }

            // 先取出消费者，再取出消息，防止由于消费者为空，导致取出的消息丢失
            // （虽然有应答机制，消息无应答会重新推送）

            // 3.调用消费者消息处理回调
            consumer->_callback(consumer->_tag, msgp->mutable_payload()->mutable_properties(), msgp->payload().body());
            // 4.消息推送后，根据消费者自动应答标志
            //  自动：直接应答该消息并删除
            //  手动：需要收到对该消息应答，才能删除
            if (consumer->_auto_ack)
                _host->basicAck(qname, msgp->payload().properties().id());
        }
        void consumerCallback(const string &ctag, const BasicProperties *bp, const string &body) // 消费者的消息处理回调函数
        {
            // basicPublish发布消息->consume工作线程进行推送->调用消息处理回调，完成推送
            // 1.构建推送响应，推送给指定消费者
            basicConsumeResponse resp;
            resp.set_cid(_cid);
            resp.set_consumer_tag(ctag);
            if (bp)
            {
                resp.mutable_properties()->set_id(bp->id());
                resp.mutable_properties()->set_delivery_mode(bp->delivery_mode());
                resp.mutable_properties()->set_routing_key(bp->routing_key());
            }
            resp.set_body(body);
            _codec->send(_conn, resp);
        }

    private:
        string _cid; // 信道ID

        Consumer::ptr _consumer;   // 一个信道只关联一个消费者:该信道专为该消费者提供服务
        ConsumerManager::ptr _cmp; // 消费者管理句柄

        muduo::net::TcpConnectionPtr _conn; // 信道关联的连接
        ProtobufCodecPtr _codec;            // protobuf协议处理句柄

        VirtualHost::ptr _host; // 虚拟机句柄
        ThreadPool::ptr _tpool; // 工作线程池句柄
    };

    // 信道管理类
    class ChannelManager
    {
    public:
        using ptr = shared_ptr<ChannelManager>;
        ChannelManager() {}
        bool openChanel(const string &cid,
                        const ConsumerManager::ptr &cmp,
                        const TcpConnectionPtr &conn,
                        const ProtobufCodecPtr &codec,
                        const VirtualHost::ptr &host,
                        const ThreadPool::ptr &tpool) // 打开信道
        {
            unique_lock<mutex> lock(mtx);
            // 1.判断信道是否存在
            auto it = _channels.find(cid);
            if (it != _channels.end()) // 重复打开同一信道
            {
                return false;
            }

            // 2.创建信道，添加信道管理
            Channel::ptr channel = make_shared<Channel>(cid, cmp, conn, codec, host, tpool);
            _channels.insert(make_pair(cid, channel));
            return true;
        }
        void closeChannel(const string &cid) // 关闭信道
        {
            unique_lock<mutex> lock(mtx);
            _channels.erase(cid);
        }
        Channel::ptr getChannel(const string &cid) // 获取信道
        {
            unique_lock<mutex> lock(mtx);
            auto it = _channels.find(cid);
            if (it == _channels.end())
            {
                return Channel::ptr();
            }
            return it->second;
        }

    private:
        mutex mtx;                                     // 加锁保护
        unordered_map<string, Channel::ptr> _channels; // 根据信道id找到对应信道
    };
}
#endif
